<?php

namespace app\amqp\producer;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\facade\Log;

class TestProducer
{
    public function publish(string $exchangeName, string $routingKey, string $data) :void
    {
        $amqpConfig = config('amqp');

        $conn = new AMQPStreamConnection(
            host: $amqpConfig['host'],
            port: $amqpConfig['port'],
            user: $amqpConfig['user'],
            password: $amqpConfig['password'],
            vhost: $amqpConfig['vhost']
        );

        $channel = $conn->channel();

        // 开启消息生产确认机制,可以监听从是否将消息发送到了交换机
        $channel->confirm_select();

//         消息成功到达交换机时触发
        $channel->set_ack_handler(function (AMQPMessage $message) {
//            var_dump(2);
            Log::info($message->body);
        });

//         消息未成功到达交换机时触发
        $channel->set_nack_handler(function (AMQPMessage $message) {
            Log::info($message->body);
        });

//         设置 return 回调函数
//         当消息从交换机路由到Queue失败才会回调这个方法
//         可以监听当发送的消息 路由key找不到 可以触发此事件
        $channel->set_return_listener(function ($replyCode, $replyText, $exchange, $routingKey, AMQPMessage $message) {
            var_dump(2);
            echo "Message returned with code {$replyCode}: {$replyText}\n";
//             在这里处理无法路由的消息
        });

        /*
        * 创建AMQP消息类型
        * $messageBody:消息体
        * delivery_mode 消息是否持久化
        *      AMQPMessage::DELIVERY_MODE_NON_PERSISTENT = 1; 不持久化
        *      AMQPMessage::DELIVERY_MODE_PERSISTENT = 2; 持久化
        */
        $message = new AMQPMessage(
            body: $data,
            properties: ['content_type' => 'application/json', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]
        );

        $channel->basic_publish(
            msg: $message,
            exchange: $exchangeName,
            routing_key: $routingKey,
            mandatory: true // 开启return消息机制，可以监听从交换机到队列是否到达
        );

        // 必须将此代码写在发布消息之后，否则不会触发回调
        // 下面与上面的区别是 下面可以监听return消息机制
        $channel->wait_for_pending_acks();
        //$channel->wait_for_pending_acks_returns();

        $channel->close();
        $conn->close();
    }
}
